// package com.jourwon.spring.boot.config;
//
// import lombok.extern.slf4j.Slf4j;
// import org.apache.kafka.clients.CommonClientConfigs;
// import org.apache.kafka.clients.admin.AdminClient;
// import org.apache.kafka.clients.admin.AdminClientConfig;
// import org.apache.kafka.clients.admin.KafkaAdminClient;
// import org.apache.kafka.clients.producer.ProducerConfig;
// import org.apache.kafka.common.config.SaslConfigs;
// import org.apache.kafka.common.serialization.StringSerializer;
// import org.springframework.beans.factory.annotation.Value;
// import org.springframework.context.annotation.Bean;
// import org.springframework.context.annotation.Configuration;
// import org.springframework.kafka.annotation.EnableKafka;
// import org.springframework.kafka.core.DefaultKafkaProducerFactory;
// import org.springframework.kafka.core.KafkaTemplate;
// import org.springframework.kafka.core.ProducerFactory;
//
// import java.util.HashMap;
// import java.util.Map;
//
// /**
//  * Kafka生产者配置
//  *
//  * @author JourWon
//  * @date 2022/4/21
//  */
// @Slf4j
// @EnableKafka
// @Configuration
// public class KafkaProducerConfig /*implements DisposableBean*/ {
//
//     /**
//      * kafka服务的ip端口,集群用逗号分隔的主机:端口对列表
//      */
//     @Value("${spring.kafka.bootstrap-servers}")
//     private String bootstrapServers;
//
//     /**
//      * 以下为kafka生产者配置
//      */
//     @Value("${spring.kafka.producer.buffer-memory}")
//     private int bufferMemory;
//
//     @Value("${spring.kafka.producer.batch-size}")
//     private int batchSize;
//
//     @Value("${spring.kafka.producer.retries}")
//     private int retries;
//
//     @Value("${spring.kafka.producer.acks}")
//     private String acks;
//
//     @Value("${spring.kafka.producer.properties.max.request.size}")
//     private int maxRequestSize;
//
//     @Value("${spring.kafka.producer.properties.linger.ms}")
//     private int lingerMs;
//
//     @Value("${spring.kafka.producer.properties.sasl.mechanism:}")
//     private String saslMechanism;
//
//     @Value("${spring.kafka.producer.properties.security.protocol:}")
//     private String securityProtocol;
//
//     @Value("${spring.kafka.producer.properties.sasl.jaas.config:}")
//     private String saslJaasConfig;
//
//     /**
//      * 以下为kafka鉴权配置
//      */
//     // @Value("${spring.kafka.jaas.login-module:}")
//     // private String jaasLoginModule;
//
//     @Value("${spring.kafka.jaas.enabled:false}")
//     private Boolean enabled;
//
//     // @Value("${spring.kafka.jaas.options.username:}")
//     // private String jaasUsername;
//     //
//     // @Value("${spring.kafka.jaas.options.password:}")
//     // private String jaasPassword;
//
//     // @PostConstruct
//     // public void initJassFile() throws IOException {
//     //     if (Boolean.FALSE.equals(enabled)) {
//     //         return;
//     //     }
//     //
//     //     File file = FileUtils.getFile(defaultJaasFileName());
//     //     file.deleteOnExit();
//     //
//     //     String jaasFileContent = "KafkaClient {\n" +
//     //             "\t" + jaasLoginModule + " required\n" +
//     //             "\tusername=\"" + jaasUsername + "\"\n" +
//     //             "\tpassword=\"" + jaasPassword + "\";\n" +
//     //             "};";
//     //
//     //     if (!file.exists()) {
//     //         FileUtils.writeStringToFile(file, jaasFileContent, StandardCharsets.UTF_8);
//     //     }
//     //
//     //     System.setProperty(JaasUtils.JAVA_LOGIN_CONFIG_PARAM, file.getAbsolutePath());
//     //
//     //     log.info("KAFKA JASS FILE={},CONFIG={}", file.getAbsolutePath(), jaasFileContent);
//     // }
//     //
//     // private String defaultJaasFileName() {
//     //     return System.getProperty("user.dir") + File.separator + "kafka_client_jaas.conf";
//     // }
//     //
//     // @Override
//     // public void destroy() {
//     //     if (StringUtils.isEmpty(jaasUsername)) {
//     //         return;
//     //     }
//     //     System.clearProperty(JaasUtils.JAVA_LOGIN_CONFIG_PARAM);
//     // }
//
//     private Map<String, Object> producerConfigs() {
//         Map<String, Object> propsMap = new HashMap<>(16);
//         propsMap.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
//         propsMap.put(ProducerConfig.BUFFER_MEMORY_CONFIG, bufferMemory);
//         propsMap.put(ProducerConfig.BATCH_SIZE_CONFIG, batchSize);
//         propsMap.put(ProducerConfig.RETRIES_CONFIG, retries);
//         propsMap.put(ProducerConfig.ACKS_CONFIG, acks);
//         propsMap.put(ProducerConfig.MAX_REQUEST_SIZE_CONFIG, maxRequestSize);
//         propsMap.put(ProducerConfig.LINGER_MS_CONFIG, lingerMs);
//         propsMap.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
//         propsMap.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
//
//         if (Boolean.TRUE.equals(enabled)) {
//             propsMap.put(SaslConfigs.SASL_MECHANISM, saslMechanism);
//             propsMap.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, securityProtocol);
//             propsMap.put(SaslConfigs.SASL_JAAS_CONFIG, saslJaasConfig);
//         }
//
//         return propsMap;
//     }
//
//     @Bean
//     public ProducerFactory producerFactory() {
//         return new DefaultKafkaProducerFactory<>(producerConfigs());
//     }
//
//     @Bean
//     public KafkaTemplate<String, String> kafkaTemplate(ProducerFactory producerFactory) {
//         return new KafkaTemplate<String, String>(producerFactory);
//     }
//
//     @Bean
//     public AdminClient kafkaAdminClient() {
//         return KafkaAdminClient.create(producerConfigs());
//     }
//
// }
